/*
 * Copyright (c) 2005 - 2010 CAS Dev Team
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 4. Neither the name of the CAS Dev. Team nor the names of its contributors
 *    may be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 *      ASMemcached.cpp
 *
 * $CAS$
 */

#include "ASMemcached.hpp"
#include <ASException.hpp>

#include <stdarg.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/param.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sysexits.h>
#include <unistd.h>
#include <syslog.h>

#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/poll.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <sys/uio.h>

#define CRLF "\r\n"
#define END_TRAILER          "END" CRLF
#define STORED_TRAILER       "STORED" CRLF
#define NOT_STORED_TRAILER   "NOT STORED" CRLF
#define CLIENT_ERROR_TRAILER "CLIENT_ERROR"
#define EXISTS_TRAILER       "EXISTS" CRLF
#define DELETED_TRAILER      "DELETED" CRLF
#define NOT_FOUND_TRAILER    "NOT_FOUND" CRLF
#define PING                 "XYI" CRLF

#define HEADER_BUFFER_SIZE 512
#define READ_BUFFER_SIZE   32768

namespace CAS_MOD
{

static CCHAR_P StrNStr(CCHAR_P szBig, CCHAR_P szLittle, UINT_32 iMaxLen)
{
	const UINT_32 iLittleLen = strlen(szLittle);
	if (iLittleLen == 0) { return szBig; }

	while (*szBig != '\0')
	{
		if (strncmp(szBig, szLittle, iLittleLen) == 0) { return szBig; }

		--iMaxLen;
		if (iMaxLen == 0) { break; }
		++szBig;
	}
return NULL;
}

//
// Resolve server host name
//
static INT_32 ResolveHostname(const STLW::string & sHost, STLW::string & sIP)
{
	// Is this IP address?
	if (inet_addr(sHost.c_str()) != -1)
	{
		sIP = sHost;
		return 0;
	}

	struct hostent * oHostEnt = gethostbyname(sHost.c_str());
	if (oHostEnt == NULL) { return -1; }

	sIP = inet_ntoa(*(struct in_addr *)(oHostEnt -> h_addr));

return 0;
}

//
// Condtructor
//
Memcached::Memcached(const STLW::string  & sIHost,
                     INT_32                iIPort,
                     INT_32                iIProto,
                     INT_32                iIConnectTimeout,
                     INT_32                iIIOTimeout): sHost(sIHost),
                                                         iPort(iIPort),
                                                         iProto(iIProto),
                                                         iConnectTimeout(iIConnectTimeout),
                                                         iIOTimeout(iIIOTimeout),
                                                         iSocket(-1),
                                                         iError(0)
{
	;;
}

//
// Get value
//
Memcached::eQueryResult Memcached::Get(const STLW::string & sKey, STLW::string & sValue, UINT_64 & iFlags)
{
	UINT_64 iCASUnique = 0;

return MemcachedGet("get", sKey, sValue, iFlags, iCASUnique);
}

//
// Get value
//
Memcached::eQueryResult Memcached::Get(const STLW::string & sKey, STLW::string & sValue)
{
	UINT_64 iFlags     = 0;
	UINT_64 iCASUnique = 0;

return MemcachedGet("get", sKey, sValue, iFlags, iCASUnique);
}


//
// Get value
//
Memcached::eQueryResult Memcached::Gets(const STLW::string & sKey, STLW::string & sValue, UINT_64 & iFlags, UINT_64 & iCASUnique)
{
	return MemcachedGet("gets", sKey, sValue, iFlags, iCASUnique);
}

//
// Get value
//
Memcached::eQueryResult Memcached::Gets(const STLW::string & sKey, STLW::string & sValue, UINT_64 & iCASUnique)
{
	UINT_64 iFlags = 0;

return MemcachedGet("gets", sKey, sValue, iFlags, iCASUnique);
}


//
// Set value
//
Memcached::eQueryResult Memcached::Set(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iFlags, INT_32 iExpires)
{
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("set", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Set value
//
Memcached::eQueryResult Memcached::Set(const STLW::string & sKey, const STLW::string & sValue, INT_32 iExpires)
{
	const UINT_64 iFlags     = 0;
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("set", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Add value
//
Memcached::eQueryResult Memcached::Add(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iFlags, INT_32 iExpires)
{
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("add", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Add value
//
Memcached::eQueryResult Memcached::Add(const STLW::string & sKey, const STLW::string & sValue, INT_32 iExpires)
{
	const UINT_64 iFlags     = 0;
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("add", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Increment value
//
Memcached::eQueryResult Memcached::Incr(const STLW::string & sKey, INT_64 & iValue)
{
	return MemcachedMath("incr", sKey, iValue);
}

//
// Decrement value
//
Memcached::eQueryResult Memcached::Decr(const STLW::string & sKey, INT_64 & iValue)
{
	return MemcachedMath("decr", sKey, iValue);
}

//
// Replace value
//
Memcached::eQueryResult Memcached::Replace(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iFlags, INT_32 iExpires)
{
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("replace", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Replace value
//
Memcached::eQueryResult Memcached::Replace(const STLW::string & sKey, const STLW::string & sValue, INT_32 iExpires)
{
	const UINT_64 iFlags     = 0;
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("replace", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Append value
//
Memcached::eQueryResult Memcached::Append(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iFlags, INT_32 iExpires)
{
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("append", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Append value
//
Memcached::eQueryResult Memcached::Append(const STLW::string & sKey, const STLW::string & sValue, INT_32 iExpires)
{
	const UINT_64 iFlags     = 0;
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("append", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Prepend value
//
Memcached::eQueryResult Memcached::Prepend(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iFlags, INT_32 iExpires)
{
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("prepend", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Prepend value
//
Memcached::eQueryResult Memcached::Prepend(const STLW::string & sKey, const STLW::string & sValue, INT_32 iExpires)
{
	const UINT_64 iFlags     = 0;
	const UINT_64 iCASUnique = UINT_64(-1);

return MemcachedSet("prepend", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Delete value
//
Memcached::eQueryResult Memcached::Del(const STLW::string & sKey)
{
	if (iSocket == -1) { return NOT_CONNECTED; }

	CHAR_8 szTMPBuf[HEADER_BUFFER_SIZE];
	CHAR_8 vReadBuffer[READ_BUFFER_SIZE];
	INT_32 iReadBufferLength = READ_BUFFER_SIZE;

	INT_32 iHeaderLen = snprintf(szTMPBuf, HEADER_BUFFER_SIZE, "delete %s" CRLF, sKey.c_str());

	INT_32 iTotalWrote = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLOUT, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iWroteBytes = 0;
		do
		{
			iWroteBytes = write(iSocket, szTMPBuf + iTotalWrote, iHeaderLen - iTotalWrote);
		} while(iWroteBytes == -1 && errno == EINTR);

		if (iWroteBytes == -1 || iWroteBytes == 0) { iError = errno; Disconnect(); return CLIENT_ERROR; }

		iTotalWrote += iWroteBytes;

		if (iTotalWrote == iHeaderLen) { break; }
	}

	// Read server response
	INT_32 iResponseSize = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLIN, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iReadBytes = 0;
		do
		{
			iReadBytes = read(iSocket, vReadBuffer + iResponseSize, iReadBufferLength - iResponseSize);
		} while(iReadBytes == -1 && errno == EINTR);

		// Socket error?
		if (iReadBytes == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }
		// EOF?
		// Try to reconnect
		if (iReadBytes == 0) { iError = ENOTCONN; Reconnect(); return CLIENT_ERROR; }

		iResponseSize += iReadBytes;

		if (iResponseSize > 2 && vReadBuffer[iResponseSize - 2] == '\r' && vReadBuffer[iResponseSize - 1] == '\n')
		{
			// Parse response header
			if      (strncmp(DELETED_TRAILER,   vReadBuffer, iResponseSize) == 0) { return OK;         }
			else if (strncmp(NOT_FOUND_TRAILER, vReadBuffer, iResponseSize) == 0) { return NOT_EXISTS; }
			else
			{
				return CLIENT_ERROR;
			}
		}

		if (iResponseSize == iReadBufferLength) { Disconnect(); return PROTOCOL_ERROR; }
	}

return OK;
}

//
// CAS
//
Memcached::eQueryResult Memcached::Cas(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iCASUnique, INT_32 iExpires)
{
	const UINT_64 iFlags = 0;

return MemcachedSet("cas", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// CAS
//
Memcached::eQueryResult Memcached::Cas(const STLW::string & sKey, const STLW::string & sValue, const UINT_64 & iCASUnique, const UINT_64 & iFlags, INT_32 iExpires)
{

return MemcachedSet("cas", sKey, sValue, iFlags, iExpires, iCASUnique);
}

//
// Increment/decrement value
//
Memcached::eQueryResult Memcached::MemcachedMath(CCHAR_P sMethod, const STLW::string & sKey, INT_64 & iValue)
{
	if (iSocket == -1) { return NOT_CONNECTED; }

	CHAR_8 szTMPBuf[HEADER_BUFFER_SIZE];
	CHAR_8 vReadBuffer[READ_BUFFER_SIZE];
	INT_32 iReadBufferLength = READ_BUFFER_SIZE;

	INT_32 iHeaderLen = snprintf(szTMPBuf, HEADER_BUFFER_SIZE, "%s %s %lld" CRLF, sMethod, sKey.c_str(), (long long int)(iValue));

	INT_32 iTotalWrote = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLOUT, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iWroteBytes = 0;
		do
		{
			iWroteBytes = write(iSocket, szTMPBuf + iTotalWrote, iHeaderLen - iTotalWrote);
		} while(iWroteBytes == -1 && errno == EINTR);

		if (iWroteBytes == -1 || iWroteBytes == 0) { iError = errno; Disconnect(); return CLIENT_ERROR; }

		iTotalWrote += iWroteBytes;

		if (iTotalWrote == iHeaderLen) { break; }
	}

	// Read server response
	INT_32 iResponseSize = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLIN, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iReadBytes = 0;
		do
		{
			iReadBytes = read(iSocket, vReadBuffer + iResponseSize, iReadBufferLength - iResponseSize);
		} while(iReadBytes == -1 && errno == EINTR);

		// Socket error?
		if (iReadBytes == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }
		// EOF?
		// Try to reconnect
		if (iReadBytes == 0) { iError = ENOTCONN; Reconnect(); return CLIENT_ERROR; }

		iResponseSize += iReadBytes;

		if (iResponseSize > 2 && vReadBuffer[iResponseSize - 2] == '\r' && vReadBuffer[iResponseSize - 1] == '\n')
		{
			// Parse response header
                        if      (strncmp(NOT_FOUND_TRAILER,    vReadBuffer, iResponseSize) == 0) { return NOT_EXISTS;  }
			else if (StrNStr(vReadBuffer, CLIENT_ERROR_TRAILER, iResponseSize) == vReadBuffer)
			{
				return BAD_COMMAND;
			}
			else
			{
				CCHAR_P sHeaderStart = vReadBuffer;
				CCHAR_P sHeaderEnd   = &vReadBuffer[iResponseSize - 2];

				// Read value length
				while (sHeaderStart != sHeaderEnd &&
				       (*sHeaderStart >= '0' && *sHeaderStart <= '9'))
				{
					iValue = iValue + 10 + *sHeaderStart - '0';
					++sHeaderStart;
				}
				return OK;
			}
		}

		if (iResponseSize == iReadBufferLength) { Disconnect(); return PROTOCOL_ERROR; }
	}

return OK;
}

//
// Set value
//
Memcached::eQueryResult Memcached::MemcachedSet(CCHAR_P               sMethod,
                                                const STLW::string  & sKey,
                                                const STLW::string  & sValue,
                                                const UINT_64       & iFlags,
                                                INT_32                iExpires,
                                                const UINT_64       & iCASUnique)
{
	if (iSocket == -1) { return NOT_CONNECTED; }

	CHAR_8 szTMPBuf[HEADER_BUFFER_SIZE];
	CHAR_8 vReadBuffer[READ_BUFFER_SIZE];
	INT_32 iReadBufferLength = READ_BUFFER_SIZE;

	INT_32 iHeaderLen = 0;
	if (iCASUnique == UINT_64(-1))
	{
		iHeaderLen = snprintf(szTMPBuf, HEADER_BUFFER_SIZE, "%s %s %lu %d %lu" CRLF, sMethod, sKey.c_str(), (unsigned long int)iFlags, iExpires, (unsigned long int)sValue.size());
	}
	else
	{
		iHeaderLen = snprintf(szTMPBuf, HEADER_BUFFER_SIZE, "%s %s %lu %d %lu %lu" CRLF, sMethod, sKey.c_str(), (unsigned long int)iFlags, iExpires, (unsigned long int)sValue.size(), (unsigned long int)iCASUnique);
	}

	struct iovec oWriteVector[3];
	oWriteVector[0].iov_base = (void *)szTMPBuf;
	oWriteVector[0].iov_len  = iHeaderLen;

	oWriteVector[1].iov_base = (void *)sValue.data();
	oWriteVector[1].iov_len  = sValue.size();

	oWriteVector[2].iov_base = (void *)CRLF;
	oWriteVector[2].iov_len  = sizeof(CRLF) - 1;

	// Set blocking mode.
	// TBD: switch to non-blocking
	INT_32 iSocketFlags = fcntl(iSocket, F_GETFL, 0);
	fcntl(iSocket, F_SETFL, iSocketFlags & (~O_NONBLOCK));

	// Write data
	if (writev(iSocket, oWriteVector, 3) == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }

	// Set socket non-blocking
	fcntl(iSocket, F_SETFL, iSocketFlags);

	// Read server response
	INT_32 iResponseSize = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLIN, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iReadBytes = 0;
		do
		{
			iReadBytes = read(iSocket, vReadBuffer + iResponseSize, iReadBufferLength - iResponseSize);
		} while(iReadBytes == -1 && errno == EINTR);

		// Socket error?
		if (iReadBytes == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }
		// EOF?
		// Try to reconnect
		if (iReadBytes == 0) { iError = ENOTCONN; Reconnect(); return CLIENT_ERROR; }

		iResponseSize += iReadBytes;

		if (iResponseSize > 2 && vReadBuffer[iResponseSize - 2] == '\r' && vReadBuffer[iResponseSize - 1] == '\n')
		{
			// Parse response header
			if      (strncmp(STORED_TRAILER,     vReadBuffer, iResponseSize) == 0) { return OK;           }
			else if (strncmp(NOT_STORED_TRAILER, vReadBuffer, iResponseSize) == 0) { return CLIENT_ERROR; }
			else if (strncmp(EXISTS_TRAILER,     vReadBuffer, iResponseSize) == 0) { return EXISTS;       }
			else
			{
				return CLIENT_ERROR;
			}
		}

		if (iResponseSize == iReadBufferLength) { Disconnect(); return PROTOCOL_ERROR; }
	}

return OK;
}

//
// Set/replace/append/prepend/add value
//
Memcached::eQueryResult Memcached::MemcachedGet(CCHAR_P               sMethod,
                                                const STLW::string  & sKey,
                                                STLW::string        & sValue,
                                                UINT_64             & iFlags,
                                                UINT_64             & iCASUnique)
{
	if (iSocket == -1) { return NOT_CONNECTED; }

	CHAR_8 szTMPBuf[HEADER_BUFFER_SIZE];
	CHAR_8 vReadBuffer[READ_BUFFER_SIZE];
	INT_32 iReadBufferLength = READ_BUFFER_SIZE - 1;

	INT_32 iHeaderLen = snprintf(szTMPBuf, HEADER_BUFFER_SIZE, "%s %s" CRLF, sMethod, sKey.c_str());

	INT_32 iTotalWrote = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLOUT, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iWroteBytes = 0;
		do
		{
			iWroteBytes = write(iSocket, szTMPBuf + iTotalWrote, iHeaderLen - iTotalWrote);
		} while(iWroteBytes == -1 && errno == EINTR);

		if (iWroteBytes == -1 || iWroteBytes == 0) { iError = errno; Disconnect(); return CLIENT_ERROR; }

		iTotalWrote += iWroteBytes;

		if (iTotalWrote == iHeaderLen) { break; }
	}

	// Read response
	INT_32 iNeedRead = 0;
	INT_32 iResponseSize = 0;
	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLIN, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iReadBytes = 0;
		do
		{
			iReadBytes = read(iSocket, vReadBuffer + iResponseSize, iReadBufferLength - iResponseSize);
		} while(iReadBytes == -1 && errno == EINTR);

		// Socket error?
		if (iReadBytes == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }
		// EOF?
		// Try to reconnect
		if (iReadBytes == 0) { iError = ENOTCONN; Reconnect(); return CLIENT_ERROR; }

		// Response size
		iResponseSize += iReadBytes;
		// Key not found?
		if (iResponseSize == 5 && strncmp(vReadBuffer, END_TRAILER CRLF, iResponseSize) == 0) { sValue.erase(); return NOT_EXISTS; }

		CCHAR_P sHeaderEnd = StrNStr(vReadBuffer, CRLF, iResponseSize);
		// Parse response header
		if (sHeaderEnd != NULL)
		{
			// VALUE <key> <flags> <bytes> [<unique>]\r\n
			if (!(vReadBuffer[0] == 'V' && vReadBuffer[4] == 'E')) { Reconnect(); return PROTOCOL_ERROR; }

			// Start of header
			CCHAR_P sHeaderStart = vReadBuffer + sizeof("VALUE ") - 1;

			// Get key
			while (sHeaderStart != sHeaderEnd && *sHeaderStart != ' ') { ++sHeaderStart; }
			if (sHeaderStart == sHeaderEnd) { Reconnect(); return PROTOCOL_ERROR; }

			// Skip spaces
			while (sHeaderStart != sHeaderEnd && *sHeaderStart == ' ') { ++sHeaderStart; }
			if (sHeaderStart == sHeaderEnd) { Reconnect(); return PROTOCOL_ERROR; }

			// Read flags
			iFlags = 0;
			while (sHeaderStart != sHeaderEnd &&
			       (*sHeaderStart >= '0' && *sHeaderStart <= '9'))
			{
				iFlags = iFlags * 10 + *sHeaderStart - '0';
				++sHeaderStart;
			}
			if (sHeaderStart == sHeaderEnd) { Reconnect(); return PROTOCOL_ERROR; }

			// Skip spaces
			while (sHeaderStart != sHeaderEnd && *sHeaderStart == ' ') { ++sHeaderStart; }
			if (sHeaderStart == sHeaderEnd) { Reconnect(); return PROTOCOL_ERROR; }

			// Read value length
			while (sHeaderStart != sHeaderEnd &&
			       (*sHeaderStart >= '0' && *sHeaderStart <= '9'))
			{
				iNeedRead = iNeedRead * 10 + *sHeaderStart - '0';
				++sHeaderStart;
			}

			// Check CAS, optional
			if (sHeaderStart != sHeaderEnd)
			{
				// Skip spaces
				while (sHeaderStart != sHeaderEnd && *sHeaderStart == ' ') { ++sHeaderStart; }
				if (sHeaderStart == sHeaderEnd) { Reconnect(); return PROTOCOL_ERROR; }

				iCASUnique = 0;
				while (sHeaderStart != sHeaderEnd &&
				       (*sHeaderStart >= '0' && *sHeaderStart <= '9'))
				{
					iCASUnique = iCASUnique * 10 + *sHeaderStart - '0';
					++sHeaderStart;
				}
			}

			// Tail of response
			INT_32 iReadValueBytes = iResponseSize - (sHeaderEnd - vReadBuffer + sizeof(CRLF) - 1);
			if (iReadValueBytes > 0)
			{
				if (iReadValueBytes - sizeof(END_TRAILER) - 1 == iNeedRead)
				{
					sValue.assign(sHeaderEnd + 2, iNeedRead);
					return OK;
				}

				sValue.assign(sHeaderEnd + 2, iReadValueBytes);
				iNeedRead -= iReadValueBytes;
			}
			break;
		}

		if (iResponseSize == iReadBufferLength) { Reconnect(); return PROTOCOL_ERROR; }
	}

	iResponseSize = 0;
	INT_32 iRestTail = iNeedRead;
	iNeedRead += sizeof(CRLF END_TRAILER) - 1;
 	for(;;)
	{
		// Check timeout
		eQueryResult oRC = PollSocket(POLLIN, iIOTimeout);
		if (oRC != OK) { return oRC; }

		INT_32 iReadBytes = 0;
		do
		{
			iReadBytes = read(iSocket, vReadBuffer, iReadBufferLength < iNeedRead ? iReadBufferLength : iNeedRead);
		} while(iReadBytes == -1 && errno == EINTR);

		// Socket error?
		if (iReadBytes == -1) { iError = errno; Disconnect(); return CLIENT_ERROR; }
		// EOF?
		// Try to reconnect
		if (iReadBytes == 0) { iError = ENOTCONN; Reconnect(); return CLIENT_ERROR; }

		// Response size
		iResponseSize += iReadBytes;
		if (iRestTail >= iReadBytes)
		{
			sValue.append(vReadBuffer, iReadBytes);
			iRestTail -= iReadBytes;
		}
		else
		{
			sValue.append(vReadBuffer, iRestTail);
			iRestTail = 0;
		}

		// All Done?
		if (iResponseSize == iNeedRead) { break; }
	}

return OK;
}

//
// Poll socket
//
Memcached::eQueryResult Memcached::PollSocket(INT_32 iFlags, INT_32 iTimeout)
{
	if (iSocket == -1) { return NOT_CONNECTED; }

	struct pollfd   aFileDescriptors[1];
	int             iTMP = 0;
	socklen_t       lErrorLength = sizeof(iError);

	aFileDescriptors[0].fd      = iSocket;
	aFileDescriptors[0].revents = 0;
	aFileDescriptors[0].events  = iFlags /* | POLLHUP | POLLERR | POLLNVAL*/;

	INT_32 iRC = 0;
	// Timeout, milliseconds
	do
	{
		iRC = poll(aFileDescriptors, 1, iTimeout);
	} while(iRC == -1 && errno == EINTR);

	if (iRC < 0)  { iError = errno;     Disconnect(); return CLIENT_ERROR;   }
	if (iRC == 0) { iError = ETIMEDOUT; Disconnect(); return SERVER_TIMEOUT; }

	iRC = getsockopt(iSocket, SOL_SOCKET, SO_ERROR, &iTMP, &lErrorLength);
	if (iRC == 0 && iTMP!= 0) { iError = iTMP;  Disconnect(); return CLIENT_ERROR; }

return OK;
}

//
// Connect to memcached server
//
Memcached::eQueryResult Memcached::RealConnect()
{
	Disconnect();

	// Always resolve host name!
	if (ResolveHostname(sHost, sIP) == -1) { return CLIENT_ERROR; }

	struct sockaddr_in oSockAddr;
	memset(&oSockAddr, 0, sizeof(struct sockaddr_in));

	oSockAddr.sin_family = AF_INET;
	oSockAddr.sin_port   = htons(iPort);
	if (inet_aton(sIP.c_str(), &oSockAddr.sin_addr) == 0) { return CLIENT_ERROR; }

	iSocket = socket(PF_INET, SOCK_STREAM, 0);
	if (iSocket == -1) { return CLIENT_ERROR; }

	// Set keep-alive flag
	int iValue = 0;
	if (setsockopt(iSocket, SOL_SOCKET, SO_KEEPALIVE, &iValue, sizeof(int)) == -1)
	{
		iError = errno;
		Disconnect();
		return CLIENT_ERROR;
	}

	// Set non-blocking mode
	fcntl(iSocket, F_SETFL, fcntl(iSocket, F_GETFL, 0) | O_NONBLOCK);

	// Connect to server
	INT_32 iRC = connect(iSocket, (struct sockaddr*)&oSockAddr, sizeof(struct sockaddr_in));
	if (iRC == -1 && errno != EINPROGRESS) { iError = errno; Disconnect(); return CLIENT_ERROR; }

	// Check error
	int iTMP = 0;
	socklen_t lErrorLength = sizeof(iTMP);
	iRC = getsockopt(iSocket, SOL_SOCKET, SO_ERROR, &iTMP, &lErrorLength);
	if (iRC == 0 && iTMP != 0) { iError = iTMP; Disconnect(); return CLIENT_ERROR; }

// Wait server
return PollSocket(POLLOUT, iConnectTimeout);
}

//
// Connect to memcached server
//
Memcached::eQueryResult Memcached::Connect()
{
	return RealConnect();
}

//
// Disconnect from memcached server
//
Memcached::eQueryResult Memcached::Disconnect()
{
	if (iSocket != -1)
	{
		close(iSocket);
		iSocket = -1;
	}

return OK;
}

//
// Disconnect & Connect
//
Memcached::eQueryResult Memcached::Reconnect()
{

return Connect();
}

//
// Check connection
//
Memcached::eQueryResult Memcached::IsConnected()
{
	if (iSocket == -1) { return NOT_CONNECTED; }

return OK;
}

//
// Get Unix error code
//
INT_32 Memcached::GetError() { return iError; }

//
// A destructor
//
Memcached::~Memcached() throw()
{
	Disconnect();
}

} // namespace CAS_MOD
// End.
